package com.zte.claa.inficombo.csa.app.service.mqttproxy.mqttclient;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zte.claa.inficombo.csa.app.config.MQTTParaConfig;
import com.zte.claa.inficombo.csa.app.service.mqttproxy.IMqttProxyService;

/**
 * @author 0283000115
 * 
 * 主要逻辑：
 * 1）维护mqtt-client的长连接（断链重连由管理器统一检查和发起...）；
 * 2）监听指定的多个topic的消息；
 * 3）发布消息到相应的消息主题（topic，下行指令）；
 * 
 */
public class MqttPubSubClient {
	// 日期格式化...
	// private SimpleDateFormat sdFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	// 日志...
	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	// 待发布消息线程池...
	private int maxWaitTime = 30;
	// metedata-threadpool.
	private ThreadPoolExecutor mdataPoolExecutor = null;
	private BlockingQueue<Runnable> mdataWaitQueue = null;
	private RejectedExecutionHandler mdataRejHandler = null;
	// cmd-send-threadpool.
	private ThreadPoolExecutor cmdPoolExecutor = null;
	private BlockingQueue<Runnable> cmdWaitQueue = null;
	private RejectedExecutionHandler cmdRejHandler = null;
	
	// 定时规则调度线程池.	
	private ScheduledThreadPoolExecutor timerExecutor = null;
	
	// mqtt参数...
	private MQTTParaConfig mqttPara = null;
	private MqttClient client = null;
	private MqttClientCallBack callBack = null;
	private String clientId;
	
	// mqttproxy服务
	private IMqttProxyService mqttService = null;
	
	// 下发指令的消息主题.
	private String ctrlCmdSendTopic = null;
	
	// 监听的消息主题.
	private String[] subscribeTopics = null;
	
	/**
	 * 构建器
	 * 
	 * @param para
	 * @param topicPostfix
	 */
	public MqttPubSubClient(String clientId, String[] rptTopics, String ctrlCmdSendTopic, MQTTParaConfig para, IMqttProxyService mqttService){
		this.clientId = clientId;
		this.subscribeTopics = rptTopics;
		this.ctrlCmdSendTopic = ctrlCmdSendTopic;
		this.mqttPara = para;
		this.mqttService = mqttService;
		// 初始化线程池.
		this.initThreadPoolExecutor();	
		// 创建mqtt客户端
		this.buildMqttClient();
		// 创建到mqtt服务器的连接.
		boolean b = this.connectToSvr();
		if(b){
			logger.info("Connect to svr SUCC: svrip={}, port={}, clientId={}.", 
					this.mqttPara.getSvrip(), this.mqttPara.getSvrport(), this.clientId);
		} else {
			logger.info("Connect to svr FAIL: svrip={}, port={}, clientId={}.", 
					this.mqttPara.getSvrip(), this.mqttPara.getSvrport(), this.clientId);
		}
	}
	
	// 初始化线程池（上报监听线程池，下发命令线程池，链路监测线程池）
	private void initThreadPoolExecutor(){
		// 上报数据监听处理线程池.
		this.mdataWaitQueue = new LinkedBlockingQueue<Runnable>(this.mqttPara.getQueuesize());
		this.mdataRejHandler = new ThreadPoolExecutor.DiscardOldestPolicy();
		this.mdataPoolExecutor = new ThreadPoolExecutor(1, 1, maxWaitTime, TimeUnit.SECONDS, mdataWaitQueue, mdataRejHandler);
		
		// cmd-send-threadpool.
		this.cmdWaitQueue = new LinkedBlockingQueue<Runnable>(this.mqttPara.getQueuesize());
		this.cmdRejHandler = new ThreadPoolExecutor.DiscardOldestPolicy();
		this.cmdPoolExecutor = new ThreadPoolExecutor(1, 1, maxWaitTime, TimeUnit.SECONDS, cmdWaitQueue, cmdRejHandler);

		// 连接检查...
		this.timerExecutor = new ScheduledThreadPoolExecutor(1);
		Runnable tt = new Runnable(){
			@Override
			public void run() {
				// 定时检查各个client的connect状态...
				if(client != null && !client.isConnected()){
					connectToSvr();
				}
			}
		};
		this.timerExecutor.scheduleAtFixedRate(tt, 60, mqttPara.getBreakcheckintervalsec(), TimeUnit.SECONDS);
		logger.info("CSA-Mqtt client connect checker start with period={}.", mqttPara.getBreakcheckintervalsec());
	}
	
	/**
	 * 创建 mqtt client对象...
	 */
	private void buildMqttClient(){
		// HOST = "tcp://192.168.1.102:1613";  
		StringBuffer sb = new StringBuffer("");
		sb.append(this.mqttPara.getProtocol().trim()).append("://").append(this.mqttPara.getSvrip().trim());
		sb.append(":").append(this.mqttPara.getSvrport());
		try{
			this.client = new MqttClient(sb.toString(), this.clientId, new MemoryPersistence()); 
			this.callBack = new MqttClientCallBack(this.clientId, this);
			this.client.setCallback(this.callBack); 
		}catch(Exception e){
			logger.warn("buildMqttClient failed: clientId={}, host={}", clientId, sb.toString(), e);
		}
	}
	
	/** 
     * 用来连接服务器 
     */  
    public boolean connectToSvr() {  
        MqttConnectOptions options = new MqttConnectOptions();  
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接 .
        options.setCleanSession(true);  
        String userName = this.mqttPara.getUsername();
        String pswd = this.mqttPara.getPassword();
        if(userName == null){
        	userName = "";
        }        
        if(pswd == null){
        	pswd = "";
        }
        options.setUserName(userName.trim());  
        options.setPassword(pswd.toCharArray()); 
        // 设置超时时间  
        options.setConnectionTimeout(this.mqttPara.getConntimeoutsec());  
        // 设置会话心跳时间  
        options.setKeepAliveInterval(this.mqttPara.getKeepaliveintervalsec());  
        options.setAutomaticReconnect(true);
        try {   
        	// 连接服务器
            this.client.connect(options); 
            // 订阅上报消息主题.
            this.subscribe();
            logger.info("Connect to MQTT server OK: svrip={}, svrport={}, clientid={}",
           			this.mqttPara.getSvrip(), this.mqttPara.getSvrport(), this.mqttPara.getClientidprefix());
            return true;
        } catch (Exception e) {  
           	logger.warn("Connect to MQTT server failed: svrip={}, svrport={}, clientid={}",
           			this.mqttPara.getSvrip(), this.mqttPara.getSvrport(), this.mqttPara.getClientidprefix(), e);
           	return false;
        }  
    }  
    
    // 订阅消息主题...
    public void subscribe(){
    	if(this.client == null){
    		logger.warn("Mqtt client is null!");
    		return;
    	} 
    	if(this.subscribeTopics == null || this.subscribeTopics.length == 0){
    		logger.warn("Mqtt subscribe topics is null!");
    		return;
    	}
    	try{
    		for(String sTipic : this.subscribeTopics){
    			sTipic = sTipic.trim();
    			if(sTipic.equals("")){
    				continue;
    			}
	    		//订阅消息  
	    		this.client.unsubscribe(sTipic);
	    		this.client.subscribe(sTipic, 1, new IMqttMessageListener(){
					@Override
					public void messageArrived(String topic, MqttMessage msg) throws Exception {
						onRptMessageReceived(topic, msg);
					}    			
	    		});
	    		logger.info("subscribe topic ok: "+sTipic);
    		}
    	}catch(Exception e){
    		logger.error("subscribe to topic failed: {}", this.subscribeTopics[0], e);
    	}
    }
    
    // 关闭到服务器的连接...
    public void closeConnectToSvr(){
    	if(this.client != null){
    		try{
    			for(String sTipic : this.subscribeTopics){
    	    		this.client.unsubscribe(sTipic);
    	    		logger.info("unSubscribe topic: "+sTipic);
        		}
    			this.client.disconnect();
    		}catch(Exception e){
    			logger.warn("closeConnectToSvr failed: svrip={}, svrport={}, clientid={}", 
    					this.mqttPara.getSvrip(), this.mqttPara.getSvrport(), this.mqttPara.getClientidprefix(), e);
    		}
    	}
    }
    
    // 判断是否连接到服务器...
    public boolean isConnected(){
    	if(this.client != null){
    		return this.client.isConnected();
    	}
    	return false;
    } 
    
    // 销毁client对象...
    public void destroy(){
    	this.closeConnectToSvr();
    	if(this.client != null){
    		try {
				this.client.close(true);
			} catch (MqttException e) {
				logger.warn("Client close failed: "+e.getMessage());
			}
    	}
    	if(this.mdataPoolExecutor != null){
    		this.mdataPoolExecutor.shutdownNow();
    		this.mdataPoolExecutor = null;
    	}
    	if(this.mdataWaitQueue != null){
    		this.mdataWaitQueue.clear();
    		this.mdataWaitQueue = null;
    	}    	
    	if(this.cmdPoolExecutor != null){
    		this.cmdPoolExecutor.shutdownNow();
    		this.cmdPoolExecutor = null;
    	}
    	if(this.cmdWaitQueue != null){
    		this.cmdWaitQueue.clear();
    		this.cmdWaitQueue = null;
    	}    	
    	if(this.timerExecutor != null){
    		this.timerExecutor.shutdownNow();
    		this.timerExecutor = null;
    	}  
    }
    
    // 异步方式下发命令.
    public void sendCmdData(String data) throws Exception {
    	this.publishMessageData(this.ctrlCmdSendTopic, data);
    }
 
    /** 
     * 发布下发命令消息.
     * 
     * @param projectId 
     * @param data 
     * @throws Exception 
     */  
    public void publishMessageData(String topicName, String data) throws Exception { 
    	if(client == null){
    		logger.warn("MQTT Client is null: topic={}, data={}", topicName, data.toString());
    		return;
    	}
    	this.cmdPoolExecutor.execute(new Runnable(){
			@Override
			public void run() {
				try{			
					// 发布数据...
			    	MqttTopic topic = client.getTopic(topicName);
			    	MqttMessage message = new MqttMessage();
			    	//保证消息能到达1次
			    	message.setQos(1);  	
			    	//服务端保留最新消息.
			        message.setRetained(true); 
			        message.setPayload(data.getBytes(mqttPara.getCharset()));  
			        topic.publish(message);  
			        logger.info("Message is published: topic={}, data={}.", topicName, data);
			        // MqttDeliveryToken token = topic.publish(message);  
			        // token.waitForCompletion();  
			        // logger.info("message is published completely: " + token.isComplete()); 
				}catch(Exception e){
					logger.error("Message publish failed: topic={}, data={}.", topicName, data.toString(), e);
				}
			}
    	});
    }
    
    // 接收到上报消息后，提交消息队列...
    public void onRptMessageReceived(String topic, MqttMessage msg){
    	this.mdataPoolExecutor.execute(new Runnable(){
			@Override
			public void run() {
				// 处理接收到的msg消息... msg.getPayload(); 
				logger.debug("Receive rpt msg from topic: {}, msg={}", topic, msg);
				mqttService.onDataReport(topic, msg);
			}
    	});
    }

	public void setReportTopics(String...reportTopics) {
		this.subscribeTopics = reportTopics;
		// 重新订阅...
		this.subscribe();
	}

}
